package edu.bupt.threadpool;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;

import org.bson.NewBSONDecoder;
import org.omg.CORBA.PRIVATE_MEMBER;

import com.mongodb.util.ThreadPool;

public class DataProcessor {

	/**
	 * @param args
	 */
	private ThreadPoolExecutor threadpool = (ThreadPoolExecutor) Executors.newCachedThreadPool();
	private BlockingQueue<String> waitedBlockingQueue = new ArrayBlockingQueue<String>(1000);
	private BlockingQueue<String> finishedBlockingQueue = new ArrayBlockingQueue<String>(1000);
	
	public void process(){
		
		new Thread(new Runnable() {
			@Override
			public void run() {
				while(true){
					while(!waitedBlockingQueue.isEmpty()){
						threadpool.execute(new SegmentHandler(waitedBlockingQueue, finishedBlockingQueue));
					}
					try {
						Thread.sleep(1000);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
			}
		}).start();
		
		while(true){
			LoadPapersHandler lps = new LoadPapersHandler(waitedBlockingQueue);
			threadpool.submit(lps);
			if(!finishedBlockingQueue.isEmpty())
				threadpool.execute(new WriteHandler(finishedBlockingQueue));
			
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			System.out.println("Thread pool size : " + threadpool.getActiveCount());
			
		}
		
	}
	public static void main(String[] args) {
		new DataProcessor().process();
	}
	


}
